package example.demo.netty;

import example.demo.netty.channelHandler.ClinetBusiChannelHandler;
import example.demo.netty.channelHandler.FastJsonDecoder;
import example.demo.netty.channelHandler.FastJsonEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.Exchanger;
import java.util.concurrent.TimeUnit;

/**
 *
 */
@Component
@Slf4j
public class NettyClientStart implements InitializingBean {
    /**
     * netty-client去发请求给netty-server，netty-server存储反馈的消息。
     */
    public final static Exchanger<JsonMessage> sendAndReceiveMessage = new Exchanger<>();

    @Value("${netty.server.host}")
    String nettyHost;
    @Value("${netty.server.port}")
    Integer nettyPort;

    private Map<String, Channel> channels; // <ip:port, channel>
    private Bootstrap bootstrap;
    /**
     * 借助这个发送消息给netty-server。dubbo也是拿这个发送message的，只不过被包装了一下org.apache.dubbo.remoting.transport.netty4.NettyChannel
     */
    private Channel channel;
    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;


    public Channel getChannel() {
        return channel;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        doOpen();
    }

    protected void doOpen() {
        NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup(
                Math.min(Runtime.getRuntime().availableProcessors() + 1, 32),
                new DefaultThreadFactory("NettyClientWorker", true));
//        final NettyClientHandler nettyClientHandler = new NettyClientHandler(getUrl(), this);
        bootstrap = new Bootstrap();
        bootstrap.group(nioEventLoopGroup)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                //.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getTimeout())
                .channel(NioSocketChannel.class);
        bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                /**<? 出站事件，channel.writeAndFlush(JsonMessage对象)--序列化--添加消息长度*/
                /**<? 入站事件，剥离消息长度--反序列化--业务handler*/
                /*剥离接收到的消息的长度字段，拿到实际的消息报文的字节数组*/
                ch.pipeline().addLast("frameDecoder",
                        new LengthFieldBasedFrameDecoder(65535,
                                0,
                                2,
                                0,
                                2));
                /*给发送出去的消息增加长度字段*/
                ch.pipeline().addLast("frameEncoder",
                        new LengthFieldPrepender(2));
                //反序列化
                ch.pipeline().addLast("MessageDecoder", new FastJsonDecoder());
                //序列化
                ch.pipeline().addLast("MessageEncoder", new FastJsonEncoder());
                ch.pipeline().addLast("client-busi", new ClinetBusiChannelHandler());
            }
        });
        try {
            doConnect();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    protected void doConnect() throws Exception {
        ChannelFuture future = bootstrap.connect(new InetSocketAddress(nettyHost, nettyPort));
        boolean ret = future.awaitUninterruptibly(3000L, TimeUnit.MILLISECONDS);
        if (ret && future.isSuccess()) {
            log.info("NettyClientStart--------------------netty-client-connect-netty-server-succeed");
            Channel newChannel = future.channel();
            Channel oldChannel = NettyClientStart.this.channel; // copy reference
            if (oldChannel != null) {
                oldChannel.close();
            }
            NettyClientStart.this.channel = newChannel;
        }
    }
}
